<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/xhtml">
  <head>

    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <meta content="Cask Data, Inc." name="author" />
<meta content="Copyright © 2014-2017 Cask Data, Inc." name="copyright" />


    <meta name="git_release" content="6.1.1">
    <meta name="git_hash" content="05fbac36f9f7aadeb44f5728cea35136dbc243e5">
    <meta name="git_timestamp" content="2020-02-09 08:22:47 +0800">
    <title>Spark Programs</title>

    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />
    <link rel="stylesheet" href="../_static/pygments.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap-theme.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/bootstrap-sphinx.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-dynamicscrollspy-4.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/abixTreeList-2.css" type="text/css" />
    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />

    <script type="text/javascript">
      var DOCUMENTATION_OPTIONS = {
        URL_ROOT:    '',
        VERSION:     '6.1.1',
        COLLAPSE_INDEX: false,
        FILE_SUFFIX: '.html',
        HAS_SOURCE:  false
      };
    </script>
    <script type="text/javascript" src="../_static/jquery.js"></script>
    <script type="text/javascript" src="../_static/underscore.js"></script>
    <script type="text/javascript" src="../_static/doctools.js"></script>
    <script type="text/javascript" src="../_static/language_data.js"></script>

    <link rel="shortcut icon" href="../_static/favicon.ico"/>
    <link rel="index" title="Index" href="../genindex.html" />
    <link rel="search" title="Search" href="../search.html" />
    <link rel="top" title="Cask Data Application Platform 6.1.1 Documentation" href="../index.html" />
    <link rel="up" title="核心概念" href="index.html" />
    <link rel="next" title="Workers" href="workers.html" />
    <link rel="prev" title="Services" href="services.html" />
    <!-- block extrahead -->
    <meta charset='utf-8'>
    <meta http-equiv='X-UA-Compatible' content='IE=edge,chrome=1'>
    <meta name='viewport' content='width=device-width, initial-scale=1.0, maximum-scale=1'>
    <meta name="apple-mobile-web-app-capable" content="yes">
    <!-- block extrahead end -->

</head>
<body role="document">

<!-- block navbar -->
<div id="navbar" class="navbar navbar-inverse navbar-default navbar-fixed-top">
    <div class="container-fluid">
      <div class="row">
        <div class="navbar-header">
          <!-- .btn-navbar is used as the toggle for collapsed navbar content -->
          <a class="navbar-brand" href="../table-of-contents/../../index.html">
            <span><img alt="CDAP logo" src="../_static/cdap_logo.svg"/></span>
          </a>

          <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".nav-collapse">
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
          </button>

          <div class="pull-right">
            <div class="dropdown version-dropdown">
              <a href="#" class="dropdown-toggle" data-toggle="dropdown"
                role="button" aria-haspopup="true" aria-expanded="false">
                v 6.1.1 <span class="caret"></span>
              </a>
              <ul class="dropdown-menu">
                <li><a href="//docs.cdap.io/cdap/5.1.2/en/index.html">v 5.1.2</a></li>
                <li><a href="//docs.cdap.io/cdap/4.3.4/en/index.html">v 4.3.4</a></li>
              </ul>
            </div>
          </div>
          <form class="navbar-form navbar-right navbar-search" action="../search.html" method="get">
            <div class="form-group">
              <div class="navbar-search-image material-icons"></div>
              <input type="text" name="q" class="form-control" placeholder="  Search" />
            </div>
            <input type="hidden" name="check_keywords" value="yes" />
            <input type="hidden" name="area" value="default" />
          </form>

          <div class="collapse navbar-collapse nav-collapse navbar-right navbar-navigation">
            <ul class="nav navbar-nav"><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../index.html">简介</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link current" href="../table-of-contents/../../guides.html">手册</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../reference-manual/index.html">参考</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../faqs/index.html">帮助</a></li>
            </ul>
          </div>

        </div>
      </div>
    </div>
  </div><!-- block navbar end -->
<!-- block main content -->
<div class="main-container container">
  <div class="row"><div class="col-md-2">
      <div id="sidebar" class="bs-sidenav scrollable-y-outside" role="complementary">
<!-- theme_manual: developer-manual -->
<!-- theme_manual_highlight: guides -->
<!-- sidebar_title_link: ../table-of-contents/../../guides.html -->

  <div role="note" aria-label="manuals links"><h3><a href="../table-of-contents/../../guides.html">Guides</a></h3>

    <ul class="this-page-menu">
      <li class="toctree-l1"><a href="../table-of-contents/../../user-guide/index.html" rel="nofollow">用户手册</a>
      </li>
      <li class="toctree-l1"><b><a href="../table-of-contents/../../developer-manual/index.html" rel="nofollow">开发手册</a></b>
      <nav class="pagenav">
      <ul class="current">
<li class="toctree-l1"><a class="reference internal" href="../index.html"> 简介</a></li>
<li class="toctree-l1"><a class="reference internal" href="../getting-started/index.html"> 入门指南</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../getting-started/sandbox/index.html">CDAP Sandbox</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../getting-started/sandbox/zip.html">二进制 Zip 文件</a></li>
<li class="toctree-l3"><a class="reference internal" href="../getting-started/sandbox/zip.html#cdap-sandbox">启动和停止 CDAP Sandbox</a></li>
<li class="toctree-l3"><a class="reference internal" href="../getting-started/sandbox/virtual-machine.html">虚拟机镜像</a></li>
<li class="toctree-l3"><a class="reference internal" href="../getting-started/sandbox/docker.html">Docker 镜像</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../getting-started/quick-start.html">快速入门</a></li>
<li class="toctree-l2"><a class="reference internal" href="../getting-started/dev-env.html">搭建开发环境</a></li>
<li class="toctree-l2"><a class="reference internal" href="../getting-started/start-stop-cdap.html">启动和停止 CDAP</a></li>
<li class="toctree-l2"><a class="reference internal" href="../getting-started/building-apps.html">构建并运行应用</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../overview/index.html"> 概述</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../overview/anatomy.html"> 大数据应用剖析</a></li>
<li class="toctree-l2"><a class="reference internal" href="../overview/modes.html"> 模式和组件</a></li>
<li class="toctree-l2"><a class="reference internal" href="../overview/abstractions.html"> 核心概念</a></li>
<li class="toctree-l2"><a class="reference internal" href="../overview/interfaces.html"> 编程接口</a></li>
</ul>
</li>
<li class="toctree-l1 current"><a class="reference internal" href="index.html"> 抽象概念</a><ul class="current">
<li class="toctree-l2"><a class="reference internal" href="core.html"> Core Abstractions</a></li>
<li class="toctree-l2"><a class="reference internal" href="applications.html"> Applications</a></li>
<li class="toctree-l2"><a class="reference internal" href="datasets/index.html"> Datasets</a><ul>
<li class="toctree-l3"><a class="reference internal" href="datasets/overview.html"> Overview</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/table.html"> Table API</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/fileset.html"> FileSets</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/partitioned-fileset.html"> Partitioned FileSets</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/time-partitioned-fileset.html"> TimePartitioned FileSets</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/system-custom.html"> System and Custom Datasets</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/permissions.html"> Dataset Permissions</a></li>
<li class="toctree-l3"><a class="reference internal" href="datasets/cube.html"> Cube Dataset</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="mapreduce-programs.html"> MapReduce Programs</a></li>
<li class="toctree-l2"><a class="reference internal" href="plugins.html"> Plugins</a></li>
<li class="toctree-l2"><a class="reference internal" href="schedules.html"> Schedules</a></li>
<li class="toctree-l2"><a class="reference internal" href="secure-keys.html"> Secure Keys</a></li>
<li class="toctree-l2"><a class="reference internal" href="services.html"> Services</a></li>
<li class="toctree-l2 current"><a class="current reference internal" href="#"> Spark Programs</a></li>
<li class="toctree-l2"><a class="reference internal" href="workers.html"> Workers</a></li>
<li class="toctree-l2"><a class="reference internal" href="workflows.html"> Workflows</a></li>
<li class="toctree-l2"><a class="reference internal" href="artifacts.html"> Artifacts</a></li>
<li class="toctree-l2"><a class="reference internal" href="program-lifecycle.html"> Program Lifecycle</a></li>
<li class="toctree-l2"><a class="reference internal" href="namespaces.html"> Namespaces</a></li>
<li class="toctree-l2"><a class="reference internal" href="transaction-system.html"> Transaction System</a></li>
<li class="toctree-l2"><a class="reference internal" href="transactional-messaging-system.html"> Transactional Messaging System</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../metadata/index.html"> 元数据</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../metadata/system-metadata.html"> System Metadata</a></li>
<li class="toctree-l2"><a class="reference internal" href="../metadata/discovery-lineage.html"> Discovery and Lineage</a></li>
<li class="toctree-l2"><a class="reference internal" href="../metadata/field-lineage.html"> Field Level Lineage</a></li>
<li class="toctree-l2"><a class="reference internal" href="../metadata/audit-logging.html"> Audit Logging</a></li>
<li class="toctree-l2"><a class="reference internal" href="../metadata/metadata-ui.html"> CDAP Metadata UI</a></li>
<li class="toctree-l2"><a class="reference internal" href="../metadata/programmatic-metadata.html"> Accessing metadata programmatically</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../pipelines/index.html"> 数据流管道</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/concepts-design.html"> Concepts and Design</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/getting-started.html"> Getting Started</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/studio.html"> CDAP Studio</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/creating-pipelines.html"> Creating Pipelines</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/running-pipelines.html"> Running Pipelines</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/plugin-management.html"> Plugin Management</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/plugins/index.html"> Plugin Reference</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/actions/index.html"> Action Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/sources/index.html"> Source Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/transforms/index.html"> Transform Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/analytics/index.html"> Analytic Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/sinks/index.html"> Sink Plugins</a><ul class="simple">
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/shared-plugins/index.html"> Shared Plugins</a><ul>
<li class="toctree-l4"><a class="reference internal" href="../pipelines/plugins/shared-plugins/core.html">CoreValidator</a></li>
</ul>
</li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/plugins/post-run-plugins/index.html"> Post-run Plugins</a><ul class="simple">
</ul>
</li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/developing-pipelines.html"> Developing Pipelines</a></li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/developing-plugins/index.html"> Developing Plugins</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/developing-plugins/plugin-basics.html">Plugin Basics</a></li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/developing-plugins/creating-a-plugin.html">Creating a Plugin</a></li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/developing-plugins/presentation-plugins.html">Plugin Presentation</a></li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/developing-plugins/testing-plugins.html">Testing Plugins</a></li>
<li class="toctree-l3"><a class="reference internal" href="../pipelines/developing-plugins/packaging-plugins.html">Packaging Plugins</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../pipelines/how-cdap-pipelines-work.html"> How CDAP Pipelines Work</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../cloud-runtimes/index.html"> 云平台运行</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../cloud-runtimes/concepts/index.html"> Concepts</a></li>
<li class="toctree-l2"><a class="reference internal" href="../cloud-runtimes/provisioners/index.html"> Provisioners</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../cloud-runtimes/provisioners/gcp-dataproc.html">Google Dataproc</a></li>
<li class="toctree-l3"><a class="reference internal" href="../cloud-runtimes/provisioners/aws-emr.html">Amazon Elastic MapReduce</a></li>
<li class="toctree-l3"><a class="reference internal" href="../cloud-runtimes/provisioners/remote-hadoop.html">Remote Hadoop</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../cloud-runtimes/profiles/index.html"> Profiles</a><ul>
<li class="toctree-l3"><a class="reference internal" href="../cloud-runtimes/profiles/creating-profiles.html">Creating Profiles</a></li>
<li class="toctree-l3"><a class="reference internal" href="../cloud-runtimes/profiles/assigning-profiles.html">Assigning Profiles</a></li>
<li class="toctree-l3"><a class="reference internal" href="../cloud-runtimes/profiles/admin-controls.html">Admin Controls</a></li>
</ul>
</li>
<li class="toctree-l2"><a class="reference internal" href="../cloud-runtimes/example/index.html"> Example</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../security/index.html"> 安全</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../security/client-authentication.html">Client Authentication</a></li>
<li class="toctree-l2"><a class="reference internal" href="../security/cdap-authentication-clients-java.html">CDAP Authentication Client for Java</a></li>
<li class="toctree-l2"><a class="reference internal" href="../security/cdap-authentication-clients-python.html">CDAP Authentication Client for Python</a></li>
<li class="toctree-l2"><a class="reference internal" href="../security/custom-authentication.html">Custom Authentication</a></li>
<li class="toctree-l2"><a class="reference internal" href="../security/authorization-extensions.html">Authorization Extensions</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../testing/index.html"> 测试和调试</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../testing/testing.html"> Testing a CDAP Application</a></li>
<li class="toctree-l2"><a class="reference internal" href="../testing/debugging.html"> Debugging</a></li>
<li class="toctree-l2"><a class="reference internal" href="../testing/troubleshooting.html"> Troubleshooting</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../ingesting-tools/index.html"> 数据融合</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../ingesting-tools/cdap-stream-clients-java.html">CDAP Stream Client for Java</a></li>
<li class="toctree-l2"><a class="reference internal" href="../ingesting-tools/cdap-stream-clients-python.html">CDAP Stream Client for Python</a></li>
<li class="toctree-l2"><a class="reference internal" href="../ingesting-tools/cdap-stream-clients-ruby.html">CDAP Stream Client for Ruby</a></li>
<li class="toctree-l2"><a class="reference internal" href="../ingesting-tools/cdap-flume.html">CDAP Flume</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../data-exploration/index.html"> 数据探索</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../data-exploration/filesets.html"> Fileset Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../data-exploration/tables.html"> Table Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../data-exploration/object-mapped-tables.html"> ObjectMappedTable Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../data-exploration/custom-datasets.html"> Custom Dataset Exploration</a></li>
<li class="toctree-l2"><a class="reference internal" href="../data-exploration/hive-execution-engines.html"> Hive Execution Engines</a></li>
</ul>
</li>
<li class="toctree-l1"><a class="reference internal" href="../advanced/index.html"> 高级主题</a><ul>
<li class="toctree-l2"><a class="reference internal" href="../advanced/application-logback.html"> Application Logback</a></li>
<li class="toctree-l2"><a class="reference internal" href="../advanced/best-practices.html"> Best Practices</a></li>
<li class="toctree-l2"><a class="reference internal" href="../advanced/class-loading.html"> Class Loading</a></li>
<li class="toctree-l2"><a class="reference internal" href="../advanced/configuring-resources.html"> Configuring Program Resources</a></li>
<li class="toctree-l2"><a class="reference internal" href="../advanced/program-retry-policies.html"> Program Retry Policies</a></li>
</ul>
</li>
</ul>
</nav>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../admin-manual/index.html" rel="nofollow">管理手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../integrations/index.html" rel="nofollow">集成手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../examples-manual/index.html" rel="nofollow">最佳实践</a>
      </li>
    </ul>
  </div></div>
    </div><div class="col-md-8 content" id="main-content">
    
  <div class="section" id="spark-programs">
<span id="spark"></span><h1>Spark Programs<a class="headerlink" href="#spark-programs" title="Permalink to this headline">🔗</a></h1>
<p><em>Apache Spark</em> is used for in-memory cluster computing. It lets you load large sets of
data into memory and query them repeatedly. This makes it suitable for both iterative and
interactive programs. Similar to MapReduce, Spark can access <a class="reference internal" href="#spark-datasets"><span class="std std-ref">datasets</span></a>
as both input and output. <em>Spark programs</em> in CDAP can be written in either Java or Scala.</p>
<p>To process data using Spark, specify <code class="docutils literal notranslate"><span class="pre">addSpark()</span></code> in your application specification:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
  <span class="p">...</span>
    <span class="n">addSpark</span><span class="p">(</span><span class="k">new</span> <span class="n">WordCountProgram</span><span class="p">());</span>
</pre></div>
</div>
<p>It is recommended to implement the <code class="docutils literal notranslate"><span class="pre">Spark</span></code> interface by extending the <code class="docutils literal notranslate"><span class="pre">AbstractSpark</span></code>
class, which allows the overriding of these three methods:</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configure()</span></code></li>
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code></li>
<li><code class="docutils literal notranslate"><span class="pre">destroy()</span></code></li>
</ul>
<p>You can extend from the abstract class <code class="docutils literal notranslate"><span class="pre">AbstractSpark</span></code> to simplify the implementation:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">WordCountProgram</span> <span class="kd">extends</span> <span class="n">AbstractSpark</span> <span class="p">{</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="n">SparkSpecification</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="k">return</span> <span class="n">SparkSpecification</span><span class="p">.</span><span class="na">Builder</span><span class="p">.</span><span class="na">with</span><span class="p">()</span>
      <span class="p">.</span><span class="na">setName</span><span class="p">(</span><span class="s">&quot;WordCountProgram&quot;</span><span class="p">)</span>
      <span class="p">.</span><span class="na">setDescription</span><span class="p">(</span><span class="s">&quot;Calculates word frequency&quot;</span><span class="p">)</span>
      <span class="p">.</span><span class="na">setMainClassName</span><span class="p">(</span><span class="s">&quot;com.example.WordCounter&quot;</span><span class="p">)</span>
      <span class="p">.</span><span class="na">build</span><span class="p">();</span>
  <span class="p">}</span>
  <span class="p">...</span>
<span class="p">}</span>
</pre></div>
</div>
<p>The configure method is similar to the one found in service and MapReduce programs. It
defines the name, description, and the class containing the Spark program to be executed
by the Spark framework.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">initialize()</span></code> method is invoked at runtime, before the
Spark program is executed. Because many Spark programs do not
need this method, the <code class="docutils literal notranslate"><span class="pre">AbstractSpark</span></code> class provides a default
implementation that does nothing.</p>
<p>However, if your program requires it, you can override this method to obtain access to the
<code class="docutils literal notranslate"><span class="pre">SparkConf</span></code> configuration and use it to set the <a class="reference external" href="https://spark.apache.org/docs/1.6.1/configuration.html">Spark properties</a> for the program:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kn">import</span> <span class="nn">org.apache.spark.SparkConf</span><span class="p">;</span>
<span class="p">.</span> <span class="p">.</span> <span class="p">.</span>
<span class="nd">@Override</span>
<span class="kd">protected</span> <span class="kt">void</span> <span class="nf">initialize</span><span class="p">()</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
  <span class="n">getContext</span><span class="p">().</span><span class="na">setSparkConf</span><span class="p">(</span><span class="k">new</span> <span class="n">SparkConf</span><span class="p">().</span><span class="na">set</span><span class="p">(</span><span class="s">&quot;spark.driver.extraJavaOptions&quot;</span><span class="p">,</span> <span class="s">&quot;-XX:MaxDirectMemorySize=1024m&quot;</span><span class="p">));</span>
<span class="p">}</span>
</pre></div>
</div>
<p>The <code class="docutils literal notranslate"><span class="pre">destroy()</span></code> method is invoked after the Spark program has
finished. You could perform cleanup or send a notification of program
completion, if that was required. Like <code class="docutils literal notranslate"><span class="pre">initialize()</span></code>, since many Spark programs do not
need this method, the <code class="docutils literal notranslate"><span class="pre">AbstractSpark</span></code> class also provides a default
implementation for this method that does nothing.</p>
<div class="section" id="spark-and-resources">
<h2>Spark and Resources<a class="headerlink" href="#spark-and-resources" title="Permalink to this headline">🔗</a></h2>
<p>When a Spark program is configured, the resource requirements for both the Spark driver
processes and the Spark executor processes can be set, both in terms of the amount of
memory (in megabytes) and the number of virtual cores assigned.</p>
<p>If both the memory and the number of cores needs to be set, this can be done using:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="n">setExecutorResources</span><span class="p">(</span><span class="k">new</span> <span class="n">Resources</span><span class="p">(</span><span class="mi">1024</span><span class="p">,</span> <span class="mi">2</span><span class="p">));</span>
</pre></div>
</div>
<p>In this case, 1024 MB and two cores is assigned to each executor process.</p>
</div>
<div class="section" id="cdap-spark-program">
<h2>CDAP Spark Program<a class="headerlink" href="#cdap-spark-program" title="Permalink to this headline">🔗</a></h2>
<p>The main class being set through the <code class="docutils literal notranslate"><span class="pre">setMainClass</span></code> or <code class="docutils literal notranslate"><span class="pre">setMainClassName</span></code> method inside the <code class="docutils literal notranslate"><span class="pre">Spark.configure()</span></code>
method will be executed by the Spark framework. The main class must have one of these properties:</p>
<ol class="arabic simple">
<li>Extends from <code class="docutils literal notranslate"><span class="pre">SparkMain</span></code>, if written in Scala</li>
<li>Have a <code class="docutils literal notranslate"><span class="pre">def</span> <span class="pre">main(args:</span> <span class="pre">Array[String])</span></code> method, if written in Scala</li>
<li>Implements <code class="docutils literal notranslate"><span class="pre">JavaSparkMain</span></code>, if written in Java</li>
<li>Have a <code class="docutils literal notranslate"><span class="pre">public</span> <span class="pre">static</span> <span class="pre">void</span> <span class="pre">main(String[]</span> <span class="pre">args)</span></code> method, if written in Java</li>
</ol>
<p>A user program is responsible for creating a <code class="docutils literal notranslate"><span class="pre">SparkContext</span></code> or <code class="docutils literal notranslate"><span class="pre">JavaSparkContext</span></code> instance, either inside
the <code class="docutils literal notranslate"><span class="pre">run</span></code> methods of <code class="docutils literal notranslate"><span class="pre">SparkMain</span></code> or <code class="docutils literal notranslate"><span class="pre">JavaSparkMain</span></code>, or inside their <code class="docutils literal notranslate"><span class="pre">main</span></code> methods.</p>
</div>
<div class="section" id="cdap-sparkexecutioncontext">
<h2>CDAP SparkExecutionContext<a class="headerlink" href="#cdap-sparkexecutioncontext" title="Permalink to this headline">🔗</a></h2>
<p>CDAP provides a <code class="docutils literal notranslate"><span class="pre">SparkExecutionContext</span></code>, which is needed to access <a class="reference internal" href="#spark-datasets"><span class="std std-ref">datasets</span></a> and to
interact with CDAP services such as metrics and service discovery. It is only available to Spark programs that
are extended from either <code class="docutils literal notranslate"><span class="pre">SparkMain</span></code> or <code class="docutils literal notranslate"><span class="pre">JavaSparkMain</span></code>.</p>

<script type="text/javascript">

  $(function tabbedparsedliteral25() {
    var tabs = ['scala', 'java'];
    var mapping = {'java': 'java', 'scala': 'scala'};
    var tabSetID = 'java-scala';
    for (var i = 0; i < tabs.length; i++) {
      var tab = tabs[i];
      $("#tabbedparsedliteral25 .example-tab-" + tab).click(changeExampleTab(tab, mapping, "tabbedparsedliteral25", tabSetID));
    }
  });

</script>
<div id="tabbedparsedliteral25" class="tabbed-parsed-literal dependent-java-scala">
<ul class="tabbed-parsed-literal nav-tabs">
<li class="example-tab example-tab-scala active"><a href="#">Scala</a></li>
<li class="example-tab example-tab-java "><a href="#">Java</a></li>
</ul>

<div class="tab-contents">

<div class="tab-pane tab-pane-scala active">
<div class="code code-tab">
<div class="highlight-scala">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="k">class</span> <span class="nc">MyScalaSparkProgram</span> <span class="k">extends</span> <span class="nc">SparkMain</span> <span class="o">{</span>
  <span class="k">override</span> <span class="k">def</span> <span class="n">run</span><span class="o">(</span><span class="k">implicit</span> <span class="n">sec</span><span class="k">:</span> <span class="kt">SparkExecutionContext</span><span class="o">)</span><span class="k">:</span> <span class="kt">Unit</span> <span class="o">=</span> <span class="o">{</span>
    <span class="k">val</span> <span class="n">sc</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">SparkContext</span>
    <span class="k">val</span> <span class="nc">RDD</span><span class="o">[(</span><span class="kt">String</span><span class="p">,</span> <span class="kt">String</span><span class="o">)]</span> <span class="k">=</span> <span class="n">sc</span><span class="o">.</span><span class="n">fromDataset</span><span class="o">(</span><span class="s">&quot;mydataset&quot;</span><span class="o">)</span>
      <span class="o">...</span>
  <span class="o">}</span>
<span class="o">}</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
<div class="tab-pane tab-pane-java ">
<div class="code code-tab">
<div class="highlight-java">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyJavaSparkProgram</span> <span class="kd">implements</span> <span class="n">JavaSparkMain</span> <span class="p">{</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">JavaSparkExecutionContext</span> <span class="n">sec</span><span class="p">)</span> <span class="p">{</span>
    <span class="n">JavaSparkContext</span> <span class="n">jsc</span> <span class="o">=</span> <span class="k">new</span> <span class="n">JavaSparkContext</span><span class="p">();</span>
    <span class="n">JavaPairRDD</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">rdd</span> <span class="o">=</span> <span class="n">sec</span><span class="p">.</span><span class="na">fromDataset</span><span class="p">(</span><span class="s">&quot;mydataset&quot;</span><span class="p">);</span>
      <span class="p">...</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
</div>
</div>
</div>
<div class="section" id="spark-and-datasets">
<span id="spark-datasets"></span><h2>Spark and Datasets<a class="headerlink" href="#spark-and-datasets" title="Permalink to this headline">🔗</a></h2>
<p>Spark programs in CDAP can directly access <strong>datasets</strong> similar to the way a MapReduce can.
These programs can create Spark’s Resilient Distributed Dataset (RDD) by reading a dataset
and can also write RDD to a dataset. In Scala, implicit objects are provided for reading
and writing datasets directly through the <code class="docutils literal notranslate"><span class="pre">SparkContext</span></code> and <code class="docutils literal notranslate"><span class="pre">RDD</span></code> objects.</p>
<p>In order to access a dataset in Spark, both the key and value classes have to be
serializable. Otherwise, Spark will fail to read or write them. For example, the Table
dataset has a value type of Row, which is not serializable. An <code class="docutils literal notranslate"><span class="pre">ObjectStore</span></code> dataset can
be used, provided its classes are serializable.</p>
<ul>
<li><p class="first">Creating an RDD from a dataset:</p>

<script type="text/javascript">

  $(function tabbedparsedliteral26() {
    var tabs = ['scala', 'java'];
    var mapping = {'java': 'java', 'scala': 'scala'};
    var tabSetID = 'java-scala';
    for (var i = 0; i < tabs.length; i++) {
      var tab = tabs[i];
      $("#tabbedparsedliteral26 .example-tab-" + tab).click(changeExampleTab(tab, mapping, "tabbedparsedliteral26", tabSetID));
    }
  });

</script>
<div id="tabbedparsedliteral26" class="tabbed-parsed-literal dependent-java-scala">
<ul class="tabbed-parsed-literal nav-tabs">
<li class="example-tab example-tab-scala active"><a href="#">Scala</a></li>
<li class="example-tab example-tab-java "><a href="#">Java</a></li>
</ul>

<div class="tab-contents">

<div class="tab-pane tab-pane-scala active">
<div class="code code-tab">
<div class="highlight-scala">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="k">val</span> <span class="n">sc</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">SparkContext</span>
<span class="k">val</span> <span class="n">purchaseRDD</span> <span class="k">=</span> <span class="n">sc</span><span class="o">.</span><span class="n">readFromDataset</span><span class="o">[</span><span class="kt">Array</span><span class="o">[</span><span class="kt">Byte</span><span class="o">]</span><span class="p">,</span> <span class="kt">Purchase</span><span class="o">](</span><span class="s">&quot;purchases&quot;</span><span class="o">);</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
<div class="tab-pane tab-pane-java ">
<div class="code code-tab">
<div class="highlight-java">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="n">JavaSparkContext</span> <span class="n">jsc</span> <span class="o">=</span> <span class="k">new</span> <span class="n">JavaSparkContext</span><span class="p">();</span>
<span class="n">JavaPairRDD</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Purchase</span><span class="o">&gt;</span> <span class="n">purchaseRDD</span> <span class="o">=</span> <span class="n">sec</span><span class="p">.</span><span class="na">fromDataset</span><span class="p">(</span><span class="s">&quot;purchases&quot;</span><span class="p">);</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
</div>
</div>
</li>
<li><p class="first">Writing an RDD to a dataset:</p>

<script type="text/javascript">

  $(function tabbedparsedliteral27() {
    var tabs = ['scala', 'java'];
    var mapping = {'java': 'java', 'scala': 'scala'};
    var tabSetID = 'java-scala';
    for (var i = 0; i < tabs.length; i++) {
      var tab = tabs[i];
      $("#tabbedparsedliteral27 .example-tab-" + tab).click(changeExampleTab(tab, mapping, "tabbedparsedliteral27", tabSetID));
    }
  });

</script>
<div id="tabbedparsedliteral27" class="tabbed-parsed-literal dependent-java-scala">
<ul class="tabbed-parsed-literal nav-tabs">
<li class="example-tab example-tab-scala active"><a href="#">Scala</a></li>
<li class="example-tab example-tab-java "><a href="#">Java</a></li>
</ul>

<div class="tab-contents">

<div class="tab-pane tab-pane-scala active">
<div class="code code-tab">
<div class="highlight-scala">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="n">purchaseRDD</span><span class="o">.</span><span class="n">saveAsDataset</span><span class="o">(</span><span class="s">&quot;purchases&quot;</span><span class="o">)</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
<div class="tab-pane tab-pane-java ">
<div class="code code-tab">
<div class="highlight-java">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="n">sec</span><span class="p">.</span><span class="na">saveAsDataset</span><span class="p">(</span><span class="n">purchaseRDD</span><span class="p">,</span> <span class="s">&quot;purchases&quot;</span><span class="p">);</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
</div>
</div>
</li>
</ul>
<p><strong>Note</strong>: Spark programs can read or write to datasets in different namespaces using
<a class="reference internal" href="datasets/overview.html#cross-namespace-dataset-access"><span class="std std-ref">Cross Namespace Dataset Access</span></a> by passing a
<code class="docutils literal notranslate"><span class="pre">String</span></code> containing the <span class="xref std std-term">namespace</span> as an additional parameter before the <span class="xref std std-term">dataset</span>
name parameter. (By default, if the namespace parameter is not supplied, the namespace in
which the program runs is used.).</p>
</div>
<div class="section" id="spark-and-services">
<h2>Spark and Services<a class="headerlink" href="#spark-and-services" title="Permalink to this headline">🔗</a></h2>
<p>Spark programs in CDAP, including worker nodes, can discover Services.
Service Discovery by worker nodes ensures that if an endpoint changes during the execution of a Spark program,
due to failure or another reason, worker nodes will see the most recent endpoint.</p>
<p>Here is an example of service discovery in a Spark program:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">final</span> <span class="n">ServiceDiscoverer</span> <span class="n">serviceDiscover</span> <span class="o">=</span> <span class="n">sec</span><span class="p">.</span><span class="na">getServiceDiscoverer</span><span class="p">();</span>
<span class="n">JavaPairRDD</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span> <span class="n">ranksRaw</span> <span class="o">=</span> <span class="n">ranks</span><span class="p">.</span><span class="na">mapToPair</span><span class="p">(</span><span class="k">new</span> <span class="n">PairFunction</span><span class="o">&lt;</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Double</span><span class="o">&gt;</span><span class="p">,</span>
                                                        <span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span><span class="p">()</span> <span class="p">{</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="n">Tuple2</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span> <span class="nf">call</span><span class="p">(</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Double</span><span class="o">&gt;</span> <span class="n">tuple</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">URL</span> <span class="n">serviceURL</span> <span class="o">=</span> <span class="n">serviceDiscover</span><span class="p">.</span><span class="na">getServiceURL</span><span class="p">(</span><span class="n">SparkPageRankApp</span><span class="p">.</span><span class="na">GOOGLE_TYPE_PR_SERVICE_NAME</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">serviceURL</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">throw</span> <span class="k">new</span> <span class="n">RuntimeException</span><span class="p">(</span><span class="s">&quot;Failed to discover service: &quot;</span> <span class="o">+</span>
                                                             <span class="n">SparkPageRankApp</span><span class="p">.</span><span class="na">GOOGLE_TYPE_PR_SERVICE_NAME</span><span class="p">);</span>
    <span class="p">}</span>
    <span class="k">try</span> <span class="p">{</span>
      <span class="n">URLConnection</span> <span class="n">connection</span> <span class="o">=</span> <span class="k">new</span> <span class="n">URL</span><span class="p">(</span><span class="n">serviceURL</span><span class="p">,</span> <span class="n">String</span><span class="p">.</span><span class="na">format</span><span class="p">(</span><span class="s">&quot;transform/%s&quot;</span><span class="p">,</span>
                                                                  <span class="n">tuple</span><span class="p">.</span><span class="na">_2</span><span class="p">().</span><span class="na">toString</span><span class="p">())).</span><span class="na">openConnection</span><span class="p">();</span>
      <span class="k">try</span> <span class="p">(</span>
        <span class="n">BufferedReader</span> <span class="n">reader</span> <span class="o">=</span> <span class="k">new</span> <span class="n">BufferedReader</span><span class="p">(</span><span class="k">new</span> <span class="n">InputStreamReader</span><span class="p">(</span><span class="n">connection</span><span class="p">.</span><span class="na">getInputStream</span><span class="p">(),</span> <span class="n">Charsets</span><span class="p">.</span><span class="na">UTF_8</span><span class="p">))</span>
      <span class="p">)</span> <span class="p">{</span>
        <span class="n">String</span> <span class="n">pr</span> <span class="o">=</span> <span class="n">reader</span><span class="p">.</span><span class="na">readLine</span><span class="p">();</span>
        <span class="k">return</span> <span class="k">new</span> <span class="n">Tuple2</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span><span class="p">(</span><span class="n">tuple</span><span class="p">.</span><span class="na">_1</span><span class="p">().</span><span class="na">getBytes</span><span class="p">(</span><span class="n">Charsets</span><span class="p">.</span><span class="na">UTF_8</span><span class="p">),</span> <span class="n">Integer</span><span class="p">.</span><span class="na">parseInt</span><span class="p">(</span><span class="n">pr</span><span class="p">));</span>
      <span class="p">}</span>
    <span class="p">}</span> <span class="k">catch</span> <span class="p">(</span><span class="n">Exception</span> <span class="n">e</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">LOG</span><span class="p">.</span><span class="na">warn</span><span class="p">(</span><span class="s">&quot;Failed to read the stream for service {}&quot;</span><span class="p">,</span>
                                                          <span class="n">SparkPageRankApp</span><span class="p">.</span><span class="na">GOOGLE_PR_SERVICE</span><span class="p">,</span> <span class="n">e</span><span class="p">);</span>
      <span class="k">throw</span> <span class="n">Throwables</span><span class="p">.</span><span class="na">propagate</span><span class="p">(</span><span class="n">e</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">});</span>
</pre></div>
</div>
</div>
<div class="section" id="spark-metrics">
<h2>Spark Metrics<a class="headerlink" href="#spark-metrics" title="Permalink to this headline">🔗</a></h2>
<p>Spark programs in CDAP emit metrics, similar to a MapReduce program.
CDAP collect system metrics emitted by Spark and display them in the <strong>CDAP UI</strong>.
This helps in monitoring the progress and resources used by a Spark program.
You can also emit custom user metrics from the worker nodes of your Spark program:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">final</span> <span class="n">Metrics</span> <span class="n">sparkMetrics</span> <span class="o">=</span> <span class="n">sc</span><span class="p">.</span><span class="na">getMetrics</span><span class="p">();</span>
<span class="n">JavaPairRDD</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span> <span class="n">ranksRaw</span> <span class="o">=</span> <span class="n">ranks</span><span class="p">.</span><span class="na">mapToPair</span><span class="p">(</span><span class="k">new</span> <span class="n">PairFunction</span><span class="o">&lt;</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Double</span><span class="o">&gt;</span><span class="p">,</span>
                                                        <span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span><span class="p">()</span> <span class="p">{</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="n">Tuple2</span><span class="o">&lt;</span><span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span> <span class="nf">call</span><span class="p">(</span><span class="n">Tuple2</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Double</span><span class="o">&gt;</span> <span class="n">tuple</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">tuple</span><span class="p">.</span><span class="na">_2</span><span class="p">()</span> <span class="o">&gt;</span> <span class="mi">100</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">sparkMetrics</span><span class="p">.</span><span class="na">count</span><span class="p">(</span><span class="n">MORE_THAN_100_KEY</span><span class="p">,</span> <span class="mi">1</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">});</span>
</pre></div>
</div>
</div>
<div class="section" id="spark-in-workflows">
<h2>Spark in Workflows<a class="headerlink" href="#spark-in-workflows" title="Permalink to this headline">🔗</a></h2>
<p>Spark programs in CDAP can also be added to a <a class="reference internal" href="workflows.html#workflows"><span class="std std-ref">workflow</span></a>, similar to a
<a class="reference internal" href="mapreduce-programs.html#mapreduce"><span class="std std-ref">MapReduce</span></a>. The Spark program can get information about the workflow
through the <code class="docutils literal notranslate"><span class="pre">SparkExecutionContext.getWorkflowInfo</span></code> method.</p>
</div>
<div class="section" id="transactions-and-spark">
<span id="spark-transactions"></span><h2>Transactions and Spark<a class="headerlink" href="#transactions-and-spark" title="Permalink to this headline">🔗</a></h2>
<p>When a Spark program interacts with datasets, CDAP will automatically create a
long-running transaction that covers the Spark job execution. A Spark job refers to a
Spark action and any tasks that need to be executed to evaluate the action (see
<a class="reference external" href="https://spark.apache.org/docs/1.6.1/job-scheduling.html#scheduling-within-an-application">Spark Job Scheduling</a>
for details).</p>
<p>You can also control the transaction scope yourself explicitly. It’s useful when you want
multiple Spark actions to be committed in the same transaction. For example, in Kafka
Spark Streaming, you can persist the Kafka offsets together with the changes in the
datasets in the same transaction to obtain exactly-once processing semantics.</p>
<p>When using an <em>explicit</em> transaction, you can access a dataset directly by calling the
<code class="docutils literal notranslate"><span class="pre">getDataset()</span></code> method of the <code class="docutils literal notranslate"><span class="pre">DatasetContext</span></code> provided to the transaction. However,
the dataset acquired through <code class="docutils literal notranslate"><span class="pre">getDataset()</span></code> cannot be used through a function closure.
See the section on <a class="reference internal" href="datasets/overview.html#datasets-in-programs"><span class="std std-ref">Using Datasets in Programs</span></a> for additional
information.</p>
<p>Here is an example of using an explicit transaction in Spark:</p>

<script type="text/javascript">

  $(function tabbedparsedliteral28() {
    var tabs = ['scala', 'java'];
    var mapping = {'java': 'java', 'scala': 'scala'};
    var tabSetID = 'java-scala';
    for (var i = 0; i < tabs.length; i++) {
      var tab = tabs[i];
      $("#tabbedparsedliteral28 .example-tab-" + tab).click(changeExampleTab(tab, mapping, "tabbedparsedliteral28", tabSetID));
    }
  });

</script>
<div id="tabbedparsedliteral28" class="tabbed-parsed-literal dependent-java-scala">
<ul class="tabbed-parsed-literal nav-tabs">
<li class="example-tab example-tab-scala active"><a href="#">Scala</a></li>
<li class="example-tab example-tab-java "><a href="#">Java</a></li>
</ul>

<div class="tab-contents">

<div class="tab-pane tab-pane-scala active">
<div class="code code-tab">
<div class="highlight-scala">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="c1">// Perform multiple operations in the same transaction</span>
<span class="nc">Transaction</span> <span class="o">{</span>
  <span class="c1">// Create a standard wordcount RDD</span>
  <span class="k">val</span> <span class="n">wordCountRDD</span> <span class="k">=</span> <span class="n">sc</span><span class="o">.</span><span class="n">fromStream</span><span class="o">[</span><span class="kt">String</span><span class="o">](</span><span class="s">&quot;stream&quot;</span><span class="o">)</span>
      <span class="o">.</span><span class="n">flatMap</span><span class="o">(</span><span class="k">_</span><span class="o">.</span><span class="n">split</span><span class="o">(</span><span class="s">&quot; &quot;</span><span class="o">))</span>
      <span class="o">.</span><span class="n">map</span><span class="o">((</span><span class="k">_</span><span class="o">,</span> <span class="mi">1</span><span class="o">))</span>
      <span class="o">.</span><span class="n">reduceByKey</span><span class="o">(</span><span class="k">_</span> <span class="o">+</span> <span class="k">_</span><span class="o">)</span>

  <span class="c1">// Save those words that have count &gt; 10 to the &quot;aboveten&quot; dataset</span>
  <span class="n">wordCountRDD</span>
    <span class="o">.</span><span class="n">filter</span><span class="o">(</span><span class="k">_</span><span class="o">.</span><span class="n">_2</span> <span class="o">&gt;</span> <span class="mi">10</span><span class="o">)</span>
    <span class="o">.</span><span class="n">saveAsDataset</span><span class="o">(</span><span class="s">&quot;aboveten&quot;</span><span class="o">)</span>

  <span class="c1">// Save all wordcount to an &quot;allcounts&quot; dataset</span>
  <span class="n">wordCountRDD</span><span class="o">.</span><span class="n">saveAsDataset</span><span class="o">(</span><span class="s">&quot;allcounts&quot;</span><span class="o">)</span>

  <span class="c1">// Updates to both the &quot;aboveten&quot; and &quot;allcounts&quot; datasets will be committed within the same transaction</span>
<span class="o">}</span>

<span class="c1">// Perform RDD operations together with direct dataset access in the same transaction</span>
<span class="nc">Transaction</span><span class="o">((</span><span class="n">datasetContext</span><span class="k">:</span> <span class="kt">DatasetContext</span><span class="o">)</span> <span class="k">=&gt;</span> <span class="o">{</span>
  <span class="n">sc</span><span class="o">.</span><span class="n">fromDataset</span><span class="o">[</span><span class="kt">String</span><span class="p">,</span> <span class="kt">Int</span><span class="o">](</span><span class="s">&quot;source&quot;</span><span class="o">)</span>
    <span class="o">.</span><span class="n">saveAsDataset</span><span class="o">(</span><span class="s">&quot;sink&quot;</span><span class="o">)</span>

  <span class="k">val</span> <span class="n">table</span><span class="k">:</span> <span class="kt">Table</span> <span class="o">=</span> <span class="n">datasetContext</span><span class="o">.</span><span class="n">getDataset</span><span class="o">(</span><span class="s">&quot;copyCount&quot;</span><span class="o">)</span>
  <span class="n">table</span><span class="o">.</span><span class="n">increment</span><span class="o">(</span><span class="k">new</span> <span class="nc">Increment</span><span class="o">(</span><span class="s">&quot;source&quot;</span><span class="o">,</span> <span class="s">&quot;sink&quot;</span><span class="o">,</span> <span class="mi">1L</span><span class="o">))</span>
<span class="o">})</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
<div class="tab-pane tab-pane-java ">
<div class="code code-tab">
<div class="highlight-java">
<!-- tabbed-parsed-literal start -->
<div class="highlight"><pre><span></span><span class="nd">@Override</span>
<span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">JavaSparkExecutionContext</span> <span class="n">sec</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
  <span class="c1">// Perform RDD operations together with direct dataset access in the same transaction</span>
  <span class="n">sec</span><span class="p">.</span><span class="na">execute</span><span class="p">(</span><span class="k">new</span> <span class="n">TransactionRunnable</span><span class="p">(</span><span class="n">sec</span><span class="p">));</span>
<span class="p">}</span>

<span class="kd">static</span> <span class="kd">class</span> <span class="nc">TransactionRunnable</span> <span class="kd">implements</span> <span class="n">TxRunnable</span><span class="p">,</span> <span class="n">Serializable</span> <span class="p">{</span>

  <span class="kd">private</span> <span class="kd">final</span> <span class="n">JavaSparkExecutionContext</span> <span class="n">sec</span><span class="p">;</span>

  <span class="kd">public</span> <span class="nf">TransactionRunnable</span><span class="p">(</span><span class="n">JavaSparkExecutionContext</span> <span class="n">sec</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">this</span><span class="p">.</span><span class="na">sec</span> <span class="o">=</span> <span class="n">sec</span><span class="p">;</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">run</span><span class="p">(</span><span class="n">DatasetContext</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">JavaPairRDD</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">Integer</span><span class="o">&gt;</span> <span class="n">source</span> <span class="o">=</span> <span class="n">sec</span><span class="p">.</span><span class="na">fromDataset</span><span class="p">(</span><span class="s">&quot;source&quot;</span><span class="p">);</span>
    <span class="n">sec</span><span class="p">.</span><span class="na">saveAsDataset</span><span class="p">(</span><span class="n">source</span><span class="p">,</span> <span class="s">&quot;sink&quot;</span><span class="p">);</span>

    <span class="n">Table</span> <span class="n">table</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getDataset</span><span class="p">(</span><span class="s">&quot;copyCount&quot;</span><span class="p">);</span>
    <span class="n">table</span><span class="p">.</span><span class="na">increment</span><span class="p">(</span><span class="k">new</span> <span class="n">Increment</span><span class="p">(</span><span class="s">&quot;source&quot;</span><span class="p">,</span> <span class="s">&quot;sink&quot;</span><span class="p">,</span> <span class="mi">1L</span><span class="p">));</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
<!-- tabbed-parsed-literal end --></div>
</div>
</div>
</div>
</div>
</div>
<div class="section" id="spark-versions">
<h2>Spark Versions<a class="headerlink" href="#spark-versions" title="Permalink to this headline">🔗</a></h2>
<p>CDAP allows you to write Spark programs using either Spark1 or Spark2.
To use Spark1, you must add the <code class="docutils literal notranslate"><span class="pre">cdap-api-spark</span></code> Maven dependency:</p>
<div class="highlight-xml notranslate"><div class="highlight"><pre><span></span>. . .
<span class="nt">&lt;dependency&gt;</span>
  <span class="nt">&lt;groupId&gt;</span>io.cdap.cdap<span class="nt">&lt;/groupId&gt;</span>
  <span class="nt">&lt;artifactId&gt;</span>cdap-api-spark<span class="nt">&lt;/artifactId&gt;</span>
  <span class="nt">&lt;version&gt;</span>${cdap.version}<span class="nt">&lt;/version&gt;</span>
<span class="nt">&lt;/dependency&gt;</span>
. . .
</pre></div>
</div>
<p>To use Spark2, you must add the <code class="docutils literal notranslate"><span class="pre">cdap-api-spark2_2.11</span></code> Maven dependency:</p>
<div class="highlight-xml notranslate"><div class="highlight"><pre><span></span>. . .
<span class="nt">&lt;dependency&gt;</span>
  <span class="nt">&lt;groupId&gt;</span>io.cdap.cdap<span class="nt">&lt;/groupId&gt;</span>
  <span class="nt">&lt;artifactId&gt;</span>cdap-api-spark2_2.11<span class="nt">&lt;/artifactId&gt;</span>
  <span class="nt">&lt;version&gt;</span>${cdap.version}<span class="nt">&lt;/version&gt;</span>
<span class="nt">&lt;/dependency&gt;</span>
. . .
</pre></div>
</div>
</div>
<div class="section" id="spark-program-examples">
<h2>Spark Program Examples<a class="headerlink" href="#spark-program-examples" title="Permalink to this headline">🔗</a></h2>
<ul class="simple">
<li>For an example, the how-to guide <span class="xref std std-ref">cdap-spark-guide</span> gives a demonstration.</li>
<li>If you have problems with resolving methods when developing Spark problems in an IDE
or running Spark programs, see <a class="reference internal" href="../testing/troubleshooting.html#development-troubleshooting-spark"><span class="std std-ref">these hints</span></a>.</li>
</ul>
</div>
</div>

</div>
    <div class="col-md-2">
      <div id="right-sidebar" class="bs-sidenav scrollable-y" role="complementary">
        <div id="localtoc-scrollspy">
        </div>
      </div>
    </div></div>
</div>
<!-- block main content end -->
<!-- block footer -->
<footer class="footer">
      <div class="container">
        <div class="row">
          <div class="col-md-2 footer-left"><a title="Services" href="services.html" />Previous</a></div>
          <div class="col-md-8 footer-center"><a class="footer-tab-link" href="../table-of-contents/../../reference-manual/licenses/index.html">Copyright</a> &copy; 2014-2020 Cask Data, Inc.&bull; <a class="footer-tab-link" href="//docs.cask.co/cdap/6.1.1/cdap-docs-6.1.1-web.zip" rel="nofollow">Download</a> an archive or
<a class="footer-tab-link" href="//docs.cask.co/cdap">switch the version</a> of the documentation
          </div>
          <div class="col-md-2 footer-right"><a title="Workers" href="workers.html" />Next</a></div>
        </div>
      </div>
    </footer>
<!-- block footer end -->
<script type="text/javascript" src="../_static/bootstrap-3.3.6/js/bootstrap.min.js"></script><script type="text/javascript" src="../_static/js/bootstrap-sphinx.js"></script><script type="text/javascript" src="../_static/js/abixTreeList-2.js"></script><script type="text/javascript" src="../_static/js/cdap-dynamicscrollspy-4.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script><script type="text/javascript" src="../_static/js/copy-to-clipboard.js"></script><script type="text/javascript" src="../_static/js/jquery.mousewheel.min.js"></script><script type="text/javascript" src="../_static/js/jquery.mCustomScrollbar.js"></script><script type="text/javascript" src="../_static/js/js.cookie.js"></script><script type="text/javascript" src="../_static/js/tabbed-parsed-literal-0.2.js"></script><script type="text/javascript" src="../_static/js/cdap-onload-javascript.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script>
    <script src="https://cdap.gitee.io/docs/cdap/json-versions.js"/></script>
  </body>
</html>